// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_MAJOR_COMPACT_CHECKER_HPP_
#define _XBASE_MAJOR_COMPACT_CHECKER_HPP_

#include <vector>

#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/noncopyable.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <xthread/guard.hpp>
#include <xthread/read_write_lock.hpp>

#include "global.hpp"
#include "table_manager.hpp"
#include "kv_scanner_manager.hpp"
#include "tablet.hpp"
#include "service_provider.hpp"

namespace xbase {

	class major_compact_checker
		: private boost::noncopyable {
	public:
		major_compact_checker()
			: m_stop(false)
			, m_duration(global::the_conf->major_compact_check_interval) {}


		void run() {
			LOG(INFO, "major_compact_checker::run checker started");

			m_deadline_time = boost::get_system_time();
			m_deadline_time += m_duration;
			
			while(! m_stop)
			{
				tablet::pointer tablet = global::get_major_compact_tablet(m_deadline_time);
				if(m_stop)
				{
					break;
				}

				if(! tablet) // timeout
				{
					// check_tablet();
					// update timer
					m_deadline_time += m_duration;
				}
				else
				{
					check(tablet);
				}
			}

			LOG(INFO, "major_compact_checker::run checker quit");
		}

		void stop() {
			m_stop = true;
			global::stop_major_compact();
		}

	protected:
		void check(tablet::pointer tablet) {
			LOG(INFO, "major_compact_checker::check check " << tablet->tablet_name() << " for major compacting");

			{
				//Guard<ReadWriteLock::ReadLock> guard(tablet->lock().read_lock());
				boost::shared_lock<boost::shared_mutex> guard(tablet->lock());
				if(! tablet->is_online()) // double check, not online
				{
					LOG(DEBUG, "major_compact_checker::check tablet "
					    << tablet->tablet_name()
					    << " is not online, skip it");
					return;
				}

				tablet->start_major_compact();
			}

				// // 1. force minor compact
				// tablet->do_minor_compact_forced();

				// // 2. 再锁定并关闭该子表
				// // 将该子表锁定，不准它改变状态
				// // 并且会使其它使用者不能继续使用这个子表，比如读写、次要紧缩等等
				// {
				// 	// 不长期持有写锁，只要将状态设为关闭即可释放
				// 	Guard<ReadWriteLock::WriteLock> guard(tablet->lock().write_lock());
				
				// 	if(! tablet->is_online()) // double check, not online
				// 	{
				// 		LOG(DEBUG, "major_compact_checker::check tablet "
				// 		    << tablet->tablet_name()
				// 		    << " is not online, skip it");
				// 		continue;
				// 	}
				
				// 	if(! tablet->not_used()) // still in use
				// 	{
				// 		LOG(DEBUG, "major_compact_checker::check tablet "
				// 		    << tablet->tablet_name()
				// 		    << " still in use, skip it");
				// 		continue;
				// 	}
					
				// 	tablet->set_online(false);
				// }

				// 3. 需要紧缩，已处于关闭状态，将其紧缩
			const bool ok = tablet->major_compact_or_split();
			tablet->complete_major_compact();
			if(! ok) // 失败，将该子表下线
			{
				LOG(WARN, "major_compact_checker::check tablet "
				    << tablet->tablet_name()
				    << " major compact return false, unload it");
				// DONE: 分裂出错，不将该子表下
				// 线，而是继续设为上线状态并服
				// 务。这样处理合适吗？是不是应
				// 该将该子表下线呢？
				//
				// 根据
				// tablet::major_compact_or_split
				// 的实现，出错是由于分裂失败，
				// 而不是主要紧缩失败，所以这里
				// 还是将该子表下线比较好，交由
				// 下一个子表服务器来判断是要主
				// 要紧缩还是要分裂（即使是主要紧缩失败，也应该这样处理）
				
				// 这里是正常下线子表的流程：首先将子表对象删除，再更新根子表中的子表状态信息；若失败则退出程序。
				tablet->do_minor_compact_forced(); // 确认保存了所有数据！
				
				std::string log_message = "tablet_server: unloaded by tablet server ";
				log_message += global::the_address_info->to_string();
				log_message += " because of failure of major compacting";
				const int ret = global::the_service->unload_tablet(tablet, log_message);
				if(ret < 0) // failed
				{
					LOG(FATAL, "major_compact_checker::check unload tablet "
					    << tablet->tablet_name()
					    << " failed, quit tablet server");
					global::the_service->report_tablet_server_quit();
					assert(false);
				}
				LOG(INFO, "major_compact_checker::check tablet "
				    << tablet->tablet_name()
				    << " unloaded");
			}
			else // 成功，需要看是否分裂了
			{
				if(! tablet->is_split()) // 没分裂，则恢复上线状态，并到下一个
				{
					//Guard<ReadWriteLock::WriteLock> guard(tablet->lock().write_lock());
					// boost::unique_lock<boost::shared_mutex> guard(tablet->lock());
					// tablet->set_online(true);
					LOG(INFO, "major_compact_checker::check tablet "
					    << tablet->tablet_name()
					    << " major compact done and not split");
				}
				else // 子表分裂了，则应该将该子表下线，写分裂信息，向Master报告！
				{
					LOG(INFO, "major_compact_checker::check tablet "
					    << tablet->tablet_name()
					    << " is split, update root meta info and report to master");
					
					// 被分裂的子表状态将更新到根子表中，所以不用再设置根子表了。只将这个子表对象删除即可。
					table::pointer table = global::the_table_manager->get_table(tablet->table_name());
					if(! table)
					{
						LOG(FATAL, "major_compact_checker::check get table "
						    << tablet->table_name() << "  failed");
						
						// clean
						tablet->clean_split_path();
						
						global::the_service->report_tablet_server_quit();
						assert(false); // 不可能的事情，否则不可恢复！
					}
					else
					{
						const size_t num = table->remove_tablet(tablet);
						LOG_IF_ELSE((num != 1),
							    WARN,
							    "major_compact_checker::check remove split tablet "
							    << tablet->tablet_name() << " failed(" << num << ")",
							    INFO,
							    "major_compact_checker::check remove split tablet "
							    << tablet->tablet_name() << " ok");
						// TODO: 暂时忽略这个返回值，如果发现真有num != 1的情况，再作处理！
					}

					// 将该子表的扫描器全部关闭
					global::the_scanner_manager->close_scanner_on_tablet(tablet->tablet_name());
					
					TabletSplitReport split_report;
					const int ret = global::the_service->update_split_tablet_info(tablet,
												      split_report);
					if(ret < 0) // failed
					{
						// 子表已经分裂，不能再继续服务，写根子表又不成功，则只有将自己core掉，才能让主服务器
						// 将这个子表分配到其它子表服务器上。
						//
						// 或者在core掉之前，向根子表发送子表下线响应，若发成功，则不需要core掉，因为
						// 主服务器已经知道它被下线了。
						LOG(FATAL, "major_compact_checker::check update split tablet "
						    << tablet->tablet_name()
						    << "'s info failed, close self");
						
						// 清理目录
						tablet->clean_split_path();
						
						// 采用较好的退出方式，先向主服务器报告自己要下线，以尽快让主服务器感知到子表服务器的退出
						global::the_service->report_tablet_server_quit();
						assert(false); // TODO: 怎样有一个比较好的退出方式，不core掉
						// 或者，如何重启？
					}
					else // 写成功，向master报告
					{
						LOG(INFO, "major_compact_checker::check update split tablet "
						    << tablet->tablet_name()
						    << "'s info ok, report to master and unload it");
						response::pointer report = response::factory_type::construct();
						report->get_header().m_magic = XBASE_PROTOCOL_MAGIC;
						report->get_header().m_version = TABLET_SERVER_VERSION;
						report->get_header().m_type = TABLET_SPLIT_REPORT;
						report->get_header().m_sequence_number = 0;
						report->get_header().m_data_length = report->add_body(split_report);
						report->save_header();
						LOG_ENV(const int ret = ) global::the_master_session->write(report);
						LOG_IF_ELSE((ret < 0),
							    WARN,
							    "major_compact_checker::check send split "
							    << "report to master failed, ignore it",
							    INFO,
							    "major_compact_checker::check send split "
							    << "report to master ok");
						// 分裂报告即使没有发送成功也没问题，可以安全的忽略它
					}
				}
			}

			LOG(INFO, "major_compact_checker::run checking done");
		}

	private:
		bool m_stop;
		const boost::posix_time::seconds m_duration;
		boost::system_time m_deadline_time;
		simple_buffer m_buffer;
	};
	
} // namespace xbase

#endif	// _XBASE_MAJOR_COMPACT_CHECKER_HPP_
